//! # 数据源模块
//! 
//! 提供各种数据源的实现，包括：
//! - 内存数据源（用于测试）
//! - 消息总线数据源
//! - 文件数据源
//! - HTTP数据源
//! - 定时器数据源

use async_trait::async_trait;
use std::collections::VecDeque;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{RwLock, Mutex};
use tokio::time::{interval, timeout};
use serde::{Deserialize, Serialize};
use tracing::{debug, error, warn};

use rustcloud_bus::prelude::{Message, MessageBus, Subscriber};
use crate::core::{StreamSource, StreamOffset};
use crate::error::{StreamResult, StreamError};

/// 数据源配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SourceConfig {
    /// 数据源名称
    pub name: String,
    /// 缓冲区大小
    pub buffer_size: usize,
    /// 批处理大小
    pub batch_size: usize,
    /// 超时设置
    pub timeout: Duration,
    /// 是否自动确认消息
    pub auto_ack: bool,
}

impl Default for SourceConfig {
    fn default() -> Self {
        Self {
            name: "default-source".to_string(),
            buffer_size: 1000,
            batch_size: 10,
            timeout: Duration::from_secs(30),
            auto_ack: true,
        }
    }
}

/// 数据源统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SourceStats {
    /// 接收到的消息总数
    pub total_received: u64,
    /// 成功处理的消息数
    pub successful: u64,
    /// 失败的消息数
    pub failed: u64,
    /// 平均处理时间（毫秒）
    pub avg_processing_time_ms: f64,
    /// 最后接收时间
    pub last_received: Option<chrono::DateTime<chrono::Utc>>,
    /// 吞吐量（消息/秒）
    pub throughput: f64,
}

impl Default for SourceStats {
    fn default() -> Self {
        Self {
            total_received: 0,
            successful: 0,
            failed: 0,
            avg_processing_time_ms: 0.0,
            last_received: None,
            throughput: 0.0,
        }
    }
}

/// 内存数据源（用于测试）
#[derive(Debug)]
pub struct MemoryStreamSource {
    messages: VecDeque<Message>,
    name: String,
    stats: Arc<RwLock<SourceStats>>,
}

impl MemoryStreamSource {
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            messages: VecDeque::new(),
            name: name.into(),
            stats: Arc::new(RwLock::new(SourceStats::default())),
        }
    }
    
    /// 创建默认的内存数据源
    pub fn default() -> Self {
        Self::new("memory-source")
    }

    pub fn add_message(&mut self, message: Message) {
        self.messages.push_back(message);
    }
    
    pub fn add_messages(&mut self, messages: Vec<Message>) {
        self.messages.extend(messages);
    }
    
    /// 获取统计信息
    pub async fn get_stats(&self) -> SourceStats {
        self.stats.read().await.clone()
    }
    
    /// 更新统计信息
    async fn update_stats(&self, success: bool, processing_time: Duration) {
        let mut stats = self.stats.write().await;
        stats.total_received += 1;
        if success {
            stats.successful += 1;
        } else {
            stats.failed += 1;
        }
        
        // 更新平均处理时间
        let new_time_ms = processing_time.as_millis() as f64;
        stats.avg_processing_time_ms = 
            (stats.avg_processing_time_ms * (stats.total_received - 1) as f64 + new_time_ms) / stats.total_received as f64;
        
        stats.last_received = Some(chrono::Utc::now());
    }
}

#[async_trait]
impl StreamSource for MemoryStreamSource {
    type Item = Message;

    async fn next(&mut self) -> Option<StreamResult<Self::Item>> {
        let start = Instant::now();
        let result = self.messages.pop_front();
        let processing_time = start.elapsed();
        
        if let Some(message) = result {
            self.update_stats(true, processing_time).await;
            Some(Ok(message))
        } else {
            None
        }
    }

    fn is_exhausted(&self) -> bool {
        self.messages.is_empty()
    }

    fn name(&self) -> &str {
        &self.name
    }
}

/// 消息总线数据源
impl std::fmt::Debug for BusStreamSource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BusStreamSource")
            .field("name", &self.name)
            .field("topic_pattern", &self.topic_pattern)
            .finish()
    }
}

pub struct BusStreamSource {
    /// 数据源名称
    name: String,
    /// 消息总线
    bus: Arc<dyn MessageBus>,
    /// 订阅器
    subscriber: Box<dyn Subscriber>,
    /// 主题模式
    topic_pattern: String,
    /// 配置
    config: SourceConfig,
    /// 统计信息
    stats: Arc<RwLock<SourceStats>>,
    /// 消息缓冲区
    buffer: Arc<Mutex<VecDeque<Message>>>,
    /// 是否正在运行
    running: Arc<RwLock<bool>>,
}

impl BusStreamSource {
    /// 创建新的消息总线数据源
    pub async fn new(
        name: impl Into<String>,
        bus: Arc<dyn MessageBus>,
        topic_pattern: impl Into<String>,
        config: SourceConfig,
    ) -> StreamResult<Self> {
        let name = name.into();
        let topic_pattern = topic_pattern.into();
        
        // 创建订阅器
        let subscriber = bus.subscribe(&topic_pattern).await
            .map_err(|e| StreamError::source_error(format!("创建订阅器失败: {}", e)))?;

        Ok(Self {
            name,
            bus,
            subscriber,
            topic_pattern,
            config,
            stats: Arc::new(RwLock::new(SourceStats::default())),
            buffer: Arc::new(Mutex::new(VecDeque::with_capacity(config.buffer_size))),
            running: Arc::new(RwLock::new(false)),
        })
    }
    
    /// 使用默认配置创建消息总线数据源
    pub async fn with_defaults(
        bus: Arc<dyn MessageBus>,
        topic_pattern: impl Into<String>,
    ) -> StreamResult<Self> {
        let topic_pattern = topic_pattern.into();
        let name = format!("bus-source-{}", topic_pattern);
        Self::new(name, bus, topic_pattern, SourceConfig::default()).await
    }
    
    /// 启动消息接收
    pub async fn start(&self) -> StreamResult<()> {
        let mut running = self.running.write().await;
        if *running {
            return Ok(());
        }
        *running = true;
        drop(running);
        
        debug!("启动消息总线数据源: {}", self.name);
        
        // 启动消息接收任务
        self.start_message_receiver().await;
        
        Ok(())
    }
    
    /// 停止消息接收
    pub async fn stop(&self) -> StreamResult<()> {
        let mut running = self.running.write().await;
        *running = false;
        debug!("停止消息总线数据源: {}", self.name);
        Ok(())
    }
    
    /// 启动消息接收任务
    async fn start_message_receiver(&self) {
        let buffer = self.buffer.clone();
        let config = self.config.clone();
        let stats = self.stats.clone();
        let running = self.running.clone();
        let name = self.name.clone();
        
        tokio::spawn(async move {
            let mut ticker = interval(Duration::from_millis(10)); // 10ms间隔检查
            
            while *running.read().await {
                ticker.tick().await;
                
                // 简化实现，暂时留空
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            
            debug!("消息接收任务结束: {}", name);
        });
    }
    
    /// 获取统计信息
    pub async fn get_stats(&self) -> SourceStats {
        self.stats.read().await.clone()
    }
    
    /// 获取缓冲区大小
    pub async fn buffer_size(&self) -> usize {
        self.buffer.lock().await.len()
    }
    
    /// 清空缓冲区
    pub async fn clear_buffer(&self) {
        self.buffer.lock().await.clear();
    }
}

#[async_trait]
impl StreamSource for BusStreamSource {
    type Item = Message;

    async fn next(&mut self) -> Option<StreamResult<Self::Item>> {
        let start = Instant::now();
        
        // 从缓冲区获取消息
        let message = {
            let mut buffer = self.buffer.lock().await;
            buffer.pop_front()
        };
        
        if let Some(msg) = message {
            let processing_time = start.elapsed();
            
            // 更新统计信息
            let mut stats = self.stats.write().await;
            stats.successful += 1;
            
            // 计算吞吐量
            if let Some(last_received) = stats.last_received {
                let duration = chrono::Utc::now().signed_duration_since(last_received);
                if duration.num_seconds() > 0 {
                    stats.throughput = stats.successful as f64 / duration.num_seconds() as f64;
                }
            }
            
            // 自动确认消息（如果启用）
            if self.config.auto_ack {
                // 这里应该调用确认逻辑，暂时省略
                debug!("自动确认消息: {}", msg.id());
            }
            
            debug!("从数据源 {} 获取消息: {}", self.name, msg.id());
            Some(Ok(msg))
        } else {
            None
        }
    }

    fn is_exhausted(&self) -> bool {
        // 消息总线数据源不会耗尽，除非明确停止
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                !*self.running.read().await && self.buffer.lock().await.is_empty()
            })
        })
    }

    fn name(&self) -> &str {
        &self.name
    }
}

/// 定时器数据源
impl std::fmt::Debug for TimerStreamSource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TimerStreamSource")
            .field("name", &self.name)
            .field("interval", &self.interval)
            .finish()
    }
}

pub struct TimerStreamSource {
    name: String,
    interval: Duration,
    message_factory: Arc<dyn Fn() -> Message + Send + Sync>,
    stats: Arc<RwLock<SourceStats>>,
    running: Arc<RwLock<bool>>,
    buffer: Arc<Mutex<VecDeque<Message>>>,
}

impl TimerStreamSource {
    /// 创建定时器数据源
    pub fn new<F>(
        name: impl Into<String>,
        interval: Duration,
        message_factory: F,
    ) -> Self
    where
        F: Fn() -> Message + Send + Sync + 'static,
    {
        Self {
            name: name.into(),
            interval,
            message_factory: Arc::new(message_factory),
            stats: Arc::new(RwLock::new(SourceStats::default())),
            running: Arc::new(RwLock::new(false)),
            buffer: Arc::new(Mutex::new(VecDeque::new())),
        }
    }
    
    /// 启动定时器
    pub async fn start(&self) -> StreamResult<()> {
        let mut running = self.running.write().await;
        if *running {
            return Ok(());
        }
        *running = true;
        drop(running);
        
        debug!("启动定时器数据源: {}", self.name);
        
        // 启动定时任务
        self.start_timer_task().await;
        
        Ok(())
    }
    
    /// 停止定时器
    pub async fn stop(&self) -> StreamResult<()> {
        let mut running = self.running.write().await;
        *running = false;
        debug!("停止定时器数据源: {}", self.name);
        Ok(())
    }
    
    /// 启动定时任务
    async fn start_timer_task(&self) {
        let factory = self.message_factory.clone();
        let buffer = self.buffer.clone();
        let stats = self.stats.clone();
        let running = self.running.clone();
        let interval_duration = self.interval;
        let name = self.name.clone();
        
        tokio::spawn(async move {
            let mut ticker = interval(interval_duration);
            
            while *running.read().await {
                ticker.tick().await;
                
                // 生成消息
                let message = (factory)();
                
                // 添加到缓冲区
                buffer.lock().await.push_back(message);
                
                // 更新统计
                let mut stats = stats.write().await;
                stats.total_received += 1;
                stats.successful += 1;
                stats.last_received = Some(chrono::Utc::now());
            }
            
            debug!("定时任务结束: {}", name);
        });
    }
    
    /// 获取统计信息
    pub async fn get_stats(&self) -> SourceStats {
        self.stats.read().await.clone()
    }
}

#[async_trait]
impl StreamSource for TimerStreamSource {
    type Item = Message;

    async fn next(&mut self) -> Option<StreamResult<Self::Item>> {
        let message = self.buffer.lock().await.pop_front();
        
        if let Some(msg) = message {
            debug!("从定时器数据源 {} 获取消息: {}", self.name, msg.id());
            Some(Ok(msg))
        } else {
            None
        }
    }

    fn is_exhausted(&self) -> bool {
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                !*self.running.read().await && self.buffer.lock().await.is_empty()
            })
        })
    }

    fn name(&self) -> &str {
        &self.name
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rustcloud_bus::memory::InMemoryMessageBus;
    use rustcloud_bus::message::MessageBuilder;
    use serde_json::json;

    #[tokio::test]
    async fn test_memory_stream_source() {
        let mut source = MemoryStreamSource::new("test");
        
        let message = MessageBuilder::new()
            .with_topic("test")
            .with_payload(json!({"value": 42}))
            .build();
        
        source.add_message(message);
        
        let result = source.next().await;
        assert!(result.is_some());
        
        let stats = source.get_stats().await;
        assert_eq!(stats.total_received, 1);
        assert_eq!(stats.successful, 1);
    }

    #[tokio::test]
    async fn test_bus_stream_source() {
        let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let mut source = BusStreamSource::with_defaults(bus.clone(), "test.*").await.unwrap();
        
        // 启动数据源
        source.start().await.unwrap();
        
        // 发布消息
        let message = MessageBuilder::new()
            .topic("test.event")
            .payload(json!({"value": 123}))
            .build();
        
        bus.publish(message).await.unwrap();
        
        // 等待一段时间让消息被接收
        tokio::time::sleep(Duration::from_millis(50)).await;
        
        // 从数据源获取消息
        let result = source.next().await;
        assert!(result.is_some());
        
        // 停止数据源
        source.stop().await.unwrap();
    }

    #[tokio::test]
    async fn test_timer_stream_source() {
        let mut source = TimerStreamSource::new(
            "timer-test",
            Duration::from_millis(10),
            || {
                MessageBuilder::new()
                    .topic("timer")
                    .payload(json!({"timestamp": chrono::Utc::now().timestamp()}))
                    .build()
            },
        );
        
        // 启动定时器
        source.start().await.unwrap();
        
        // 等待生成消息
        tokio::time::sleep(Duration::from_millis(25)).await;
        
        // 获取消息
        let result = source.next().await;
        assert!(result.is_some());
        
        // 停止定时器
        source.stop().await.unwrap();
    }
}